package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo22Pagerank {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local").setAppName("pr")

    val sc = new SparkContext(conf)


    //1、读取数据
    val pagerankRDD: RDD[String] = sc.textFile("data/pagerank.txt")


    //2、将网页和出链列表切分出来
    val pageAndLinkList: RDD[(String, List[String])] = pagerankRDD.map(line => {
      val split: Array[String] = line.split("-")

      //当前网页
      val page: String = split(0)

      //出链列表
      val linkList: List[String] = split(1).split(",").toList


      (page, linkList)
    })

    //对多次使用的rdd进行缓存
    pageAndLinkList.cache()

    //网页数量
    val N: Long = pageAndLinkList.count()
    //阻尼系数
    val q = 0.85


    //3、给每一个网页一个初始的pr值 1

    val pageAndLinkListAndPr: RDD[(String, List[String], Double)] = pageAndLinkList.map {
      case (page: String, linkList: List[String]) =>
        (page, linkList, 1.0)

    }

    var inputPage: RDD[(String, List[String], Double)] = pageAndLinkListAndPr


    while (true) {


      //将每个网页的pr值平分给他的出链列表
      val pageAvgPr: RDD[(String, Double)] = inputPage.flatMap {
        case (page: String, linkList: List[String], pr: Double) =>

          //将pr平分
          val avgPr: Double = pr / linkList.length

          //将pr分给出链列表
          linkList.map(p => (p, avgPr))
      }

      //计算每个网页分到的总的pr值
      val pageSumPr: RDD[(String, Double)] = pageAvgPr.reduceByKey(_ + _)


      //将出链列表管理回去
      val pageJoin: RDD[(String, (Double, List[String]))] = pageSumPr.join(pageAndLinkList)

      //整理数据
      val currPageRDD: RDD[(String, List[String], Double)] = pageJoin.map {
        case (page: String, (pr: Double, linkList: List[String])) =>
          //增加阻尼系数
          (page, linkList, (1 - q) / N + q * pr)
      }

      //计算当前这一次所有网页和上一次所有网页pr值的差值平均值

      //当前
      val currPr: RDD[(String, Double)] = currPageRDD.map(kv => (kv._1, kv._3))
      //上一次
      val lastPr: RDD[(String, Double)] = inputPage.map(kv => (kv._1, kv._3))


      val chaJoinRDD: RDD[(String, (Double, Double))] = currPr.join(lastPr)
      val chaPr: RDD[Double] = chaJoinRDD.map {
        case (page: String, (pr1: Double, pr2: Double)) =>
          //计算差值，取绝对值
          Math.abs(pr1 - pr2)
      }

      val chaAvg: Double = chaPr.sum() / N

      currPageRDD.foreach(println)
      println(s"差值平均值：$chaAvg")

      //收敛
      if (chaAvg <= 0.001) {
        //将最后结果保存
        val resulyRDD: RDD[String] = currPageRDD.map {
          case (page: String, linkList: List[String], pr: Double) =>
            s"$page-${linkList.mkString(",")}-$pr"
        }
        resulyRDD.saveAsTextFile("data/pr")

        return
      }


      inputPage = currPageRDD


    }


  }

}
